package com.shujia.spark.core

import org.apache.spark.{SparkConf, SparkContext}
import org.apache.spark.rdd.RDD

object Demo14AggregateByKey {
  def main(args: Array[String]): Unit = {
    /**
     * aggregateByKey：类似reduceByKey，需要作用在KV格式的RDD上
     * 也会进行预聚合，只不过该算子将预聚合和聚合操作分开了
     * 如果预聚合和聚合操作一样，则可以使用reduceByKey进行简化
     */
    val conf = new SparkConf()
    conf.setMaster("local")
    conf.setAppName(this.getClass.getSimpleName.replace("$", ""))
    val sc = new SparkContext(conf)

    val sRDD: RDD[(String, Int)] = sc.parallelize(List(("001", 20), ("001", 21), ("001", 23), ("002", 21), ("002", 23)))

    sRDD
      .map(t2 => (t2._1, (t2._2.toFloat, 1)))
      /**
      * zeroValue：初始值，用于存储计算的结果，类型不限
      * seqOp：预聚合的逻辑
      * combOp：聚合的逻辑
      */
      .aggregateByKey((0.0,0))(seqOp = (u,v)=>{
      // 在分区内对每条数据进行处理
        (u._1 + v._1,u._2 + v._2)
      },combOp = (u1,u2)=>{
      // 在分区间进行聚合处理
       (u1._1 + u2._1,u1._2 + u2._2)
      })
      .map(t2=>s"${t2._1},${t2._2._1/t2._2._2}")
      .foreach(println)

    // 当seqOp和combOp一致时可以用reduceByKey进行简化
    sRDD
      .map(t2 => (t2._1, (t2._2.toDouble, 1)))
      .reduceByKey((t2_1,t2_2)=>{
        (t2_1._1 + t2_2._1,t2_1._2+t2_2._2)
      })
      .map(t2=>s"${t2._1},${t2._2._1/t2._2._2}")
      .foreach(println)


  }

}
